package com.danan.realtime.app.dwd;

import com.danan.realtime.app.BaseSqlApp;
import com.danan.realtime.common.Constant;
import com.danan.realtime.util.ResourceBundleUtil;
import com.danan.realtime.util.SinkUtil;
import com.danan.realtime.util.SourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.phoenix.schema.types.PSmallint;
import scala.collection.immutable.Stream;
import scala.tools.nsc.transform.patmat.ScalaLogic;

/**
 * @author NanHuang
 * @Date 2023/1/25
 */
public class DwdApp_2_trade_cart_add extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdApp_2_trade_cart_add().init(4002,3,"DwdApp_2_trade_cart_add");
    }

    @Override
    public void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 读取ods_db的数据
        readOdsDb(tableEnv,"DwdApp_2_trade_cart_add");
        // 2 过滤出加购事实表的数据
        filterCartInfo(tableEnv);
        // 3 查询字典表的数据
        readBaseDic(tableEnv);
        // 4 事实表join字典表（lookup join）
        extendDim(tableEnv);
        // 5 将结果写入Kafka中
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        // 创建目标表，与Kafka指定topic绑定
        tableEnv.executeSql("create table target (" +
                "   id string," +
                "   user_id string," +
                "   sku_id string," +
                "   source_id string," +
                "   source_type_id string," +
                "   source_type_name string," +
                "   sku_num int," +
                "   ts bigint" +
                ")" +
                SinkUtil.getKafkaWith(Constant.TOPIC_DWD_TRADE_CART_ADD));
        // 数据导入
        tableEnv.executeSql("insert into target select * from join_result");
    }

    private void extendDim(StreamTableEnvironment tableEnv) {
        Table joinResult = tableEnv.sqlQuery("select " +
                "id,user_id,sku_id,source_id,source_type source_type_id,dic_name source_type_name,sku_num,ts " +
                "from cart_info ci " +
                "join base_dic for system_time as of ci.pt as dic " +
                "on ci.source_type = dic.dic_code");
        tableEnv.createTemporaryView("join_result",joinResult);
    }

    private void filterCartInfo(StreamTableEnvironment tableEnv) {
        // 查询
        Table cartInfo = tableEnv.sqlQuery("select " +
                "   `data`['id'] id," +
                "   `data`['user_id'] user_id," +
                "   `data`['sku_id'] sku_id," +
                "   `data`['source_id'] source_id," +
                "   `data`['source_type'] source_type," +
                // 加购数量 = 新的数量 - 旧的数量 （流式数据，便于后续的聚合计算）
                "   if(`type`='insert',cast(`data`['sku_num'] as int),cast(`old`['sku_num'] as int) - cast(`data`['sku_num'] as int)) sku_num," +
                "   ts," +
                "   pt " +
                "from source " +
                "where `table`='cart_info' " +
                "and (" +
                "   `type`='insert' " +
                "   or (" +
                "       `type`='update' " +
                "       and `old`['sku_num'] is not null " +
                "       and cast(`old`['sku_num'] as int) > cast(`data`['sku_num'] as int)" +
                "       )" +
                ")");
        tableEnv.createTemporaryView("cart_info",cartInfo);
    }

}
